package com.itstyle.quartz.job;

import com.alibaba.fastjson.JSONObject;
import com.itstyle.quartz.entity.QuartzEntityDt;
import com.itstyle.quartz.interceptor.ReadFileThread;
import com.itstyle.quartz.kettleUtil.DateUtil;
import com.itstyle.quartz.kettleUtil.FileReadUtil;
import com.itstyle.quartz.kettleUtil.LogUtil;
import com.itstyle.quartz.kettleUtil.kettleUtil;
import com.itstyle.quartz.service.dao.ACLineDotService;
import com.itstyle.quartz.service.dao.IJobService;
import com.itstyle.quartz.entity.Result;
import org.quartz.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;

import java.io.File;
import java.io.Serializable;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.UnknownHostException;
import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author ycj
 * @version 1.0
 * @description
 * @date 2020/11/19 14:47
 */
@DisallowConcurrentExecution
public class FileJob implements Job, Serializable {

    private static final Logger logger = LoggerFactory.getLogger(kettleJob.class);

    private static final long serialVersionUID = 1L;

    @Autowired
    private static ACLineDotService acLineDotService;
    @Autowired
    private IJobService jobService;
    private JobKey jobkey;
    public FileJob(){};
    public FileJob(JobKey jobkey,IJobService jobService){this.jobkey=jobkey;this.jobService=jobService;};
    public FileJob(ACLineDotService acLineDotService){
        this.acLineDotService=acLineDotService;
    };

    @Override
//    @Transactional
    public void execute(JobExecutionContext context){
        JobDetail jobDetail = context.getJobDetail();
        JobDataMap dataMap = jobDetail.getJobDataMap();
        JobKey jobkeys = jobDetail.getKey();

        /**
         * 获取任务中保存的方法名字，动态调用方法
         */
        String methodName = dataMap.getString("jobMethodName");
        try {
            FileJob job = new FileJob(jobkeys,this.jobService);
            Method method = job.getClass().getMethod(methodName);
            method.invoke(job);
        } catch (NoSuchMethodException e) {
            logger.error("KETTLE JOB执行异常，jobkey="+jobkey.toString()+e.getMessage());
        } catch (IllegalAccessException e) {
            logger.error("KETTLE JOB执行异常，jobkey="+jobkey.toString()+e.getMessage());
        } catch (InvocationTargetException e) {
            logger.error("KETTLE JOB执行异常，jobkey="+jobkey.toString()+e.getMessage());
        }
    }

    public Result runFileJob(){
        String errorfilepath = "/home/dcloud/kettle_data_manage/kettle_manage/error";
        String transferFilePath = "/home/dcloud/kettle_data_manage/kettle_manage/bak";
//        String transferFilePath = "D:\\dianli\\k\\bak";
//        String errorfilepath = "D:\\dianli\\k\\error";
        int error = 0;
        String filename = "";
        String path=null;
        String kettlepath=null;
        org.pentaho.di.core.Result result1 = null;
        Result result = null;
        //生成一个日志主键
        String quartzid = UUID.randomUUID().toString().substring(0,10)+ DateUtil.getFormatTimeString(new Date(),"yyyyMMddHHmmss");
        // 对参数进行处理，约定为ktr不传参数，所有传参数的都为Job任务去执行
        JSONObject params = null;
        //日志状态信息给定默认值
        String message="任务执行中！";
        //任务状态
        String status="正在执行";
        //使用集群时显示主机IP
        String hostIp = "";
        try {
            hostIp = LogUtil.getLocalhostIp();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }

        if(jobkey != null){
            //根据key获取kettle对象
            List<QuartzEntityDt> list = jobService.listJobKettle(jobkey);
            if(list != null && list.size()>0){
                QuartzEntityDt quartz = list.get(0);
                //获取执行文件夹路径
                path = quartz.getFILEPATCH();
                kettlepath=quartz.getKETTLEPATCH();

                //处理日志信息
                LogUtil.saveJobKettleLogs(quartz.getFILEPATCH(),quartz.getKETTLEPATCH(),quartzid,jobkey,jobService,status,null);
                try {
                    logger.info("JOB任务开始执行：jobkey=" + jobkey.toString());

                    //获取目标文件夹下文件
                    File[] file1 = FileReadUtil.getFile(path);

                    //判断文件转移路径是否存在
                    File file2 = new File(errorfilepath);
                    if (!file2.exists()) {
                        file2.mkdir();
                    }
                    File file3 = new File(transferFilePath);
                    if (!file3.exists()) {
                        file3.mkdir();
                    }

                    //创建多线程继承对象，解析文件方法
                    ReadFileThread ReadFileThread = new ReadFileThread(path,acLineDotService,jobkey,jobService,quartzid);
                    ThreadPoolExecutor executor = new ThreadPoolExecutor(20, 100, 60L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue(500));
                    for (int i = 0; i < 11; i++) {
                        //执行线程
                        executor.execute(ReadFileThread);
                    }

                    //执行状态日志记录
                    if (kettlepath.length()>0) {
                        result1 = kettleUtil.runJob(params, kettlepath);
                        if(result1.getResult()){
                            //处理日志信息
                            status="kettle执行成功";
                            message="执行主机IP"+hostIp+"; 任务执行成功!";
                            LogUtil.updateJobKettleLogs(quartzid,jobkey,jobService,status,message);
                        }else {
                            //处理日志信息
                            status="kettle执行失败";
                            message="执行主机IP"+hostIp+"; 执行失败! 请检查执行Ketlle job是否有异常。";
                            LogUtil.updateJobKettleLogs(quartzid,jobkey,jobService,status,message);
                        }
                    }
                }catch(Exception e){
                    logger.error("KETTLE JOB执行异常，jobkey="+jobkey.toString()+e.getMessage());
                    //处理日志信息
                    status="执行异常";
                    message="执行主机IP"+hostIp+";"+e.getMessage();
                    LogUtil.updateJobKettleLogs(quartzid,jobkey,jobService,status,message);
                }
            }
        }
        return result;
    }
}
